﻿#region Header File
//
// LinkProtocol.cs - Link Layer Protocol Base Class
//
// Copyright (C) Javier Valcarce. BSD License
#endregion

#region Using Statements
using System;
using System.Diagnostics;
using System.IO;
using System.Collections.Generic;
using System.Threading;
#endregion

namespace Arq
{
    /// <summary>
    /// Link Layer ARQ Protocol. COBS Frame Encoder. 
    /// Thread-Safe: Admits multiple producers (which invoke SendPacket) and multiple consumers 
    /// (which invoke RecvPacket). Admits concurrent producers and consumers, i.e. , a producer 
    /// blocked in SendPacket() doesn't necesarilly blocks a conusmer on RecvPacket(). Both API
    /// primitives are independent.
    /// </summary>
    public abstract class LinkProtocol : Node
    {        
        #region Private Members
        // queues        
        FiniteQueue<byte[]> pktTxQ;  // queue for tx packets
        FiniteQueue<byte[]> pktRxQ;  // queue for rs packets
        FiniteQueue<byte[]> frmTxQ;  // queue for byte-stuffing encoder
        FiniteQueue<byte[]> frmRxQ;  // queue for byte-stuffing decoder

        Thread       encth;     // thread for frame encoder
        Thread       decth;     // thread for frame decoder
        ByteStuffing codec;     // byte stuffing frame encoder/decoder
        TrafficStats txstats;   // teletraffic statistics for tx
        TrafficStats rxstats;   // teletraffic statisdics for rx
        bool wakeup;
        object       evsync = new object(); // NotifyIssue and TryAgain sync
        #endregion

        #region Constructors
        public LinkProtocol(Stream istream, Stream ostream)
            : base("StopAndWait")
        {
            In  = istream;
            Out = ostream;

            RecvPacketDroped = 0;
            wakeup = false;

            codec   = new ByteStuffing(MTU + 3);
            pktTxQ  = new FiniteQueue<byte[]>(64);
            pktRxQ  = new FiniteQueue<byte[]>(64);
            frmTxQ  = new FiniteQueue<byte[]>(64);
            frmRxQ  = new FiniteQueue<byte[]>(64);            
            txstats = new TrafficStats();
            rxstats = new TrafficStats();
            //
            encth = new Thread(FrameEncoder);
            decth = new Thread(FrameDecoder);
            //
            encth.IsBackground = true;
            decth.IsBackground = true;

            encth.Start();
            decth.Start();
        }
        #endregion

        #region API for application
        /// <summary>
        /// Sends a data packet. Sends the data chunk from pkt[offset] to pkt[offset + count]. The
        /// count parameter must be equal or less than protocol MTU (Maximun Transfer Unit)
        /// </summary>
        /// <param name="pkt">Source array</param>
        /// <param name="offset">Offset in the pkt array</param>
        /// <param name="count">Number of bytes transfered, must be equal or less than protocol MTU
        /// (Maximum Transfer Unit)</param>
        public int SendPacket(byte[] pkt, int offset, /**/int count)
        {
            // copy and send
            byte[] tmp = new byte[count];
            Array.Copy(pkt, offset, tmp, 0, count);
            return SendPacket(ref tmp);
        }

        /// <summary>
        /// Sends a data packet. The parameter is set to null to ensure that the buffer is not 
        /// used by the aplication, i.e., the protocol acepts the buffer, it doesn't copy the 
        /// buffer data.
        /// </summary>
        /// <param name="pkt">Data to be sent</param>
        /// <returns>The size of the packet is all goes well or zero if en LinkError event is 
        /// raised</returns>
        public int SendPacket(ref byte[] pkt)
        {
            if (pkt.Length > MTU) throw new ArgumentException();
            if (pkt.Length ==  0) throw new ArgumentException();

            // alocate enought space (MTU + 2 bytes) to build frame format "in situ" prepending
            // headers and appending channel codes and other trailing sequences
            int result = 0;
            lock (pktTxQ)
            {
                if (pktTxQ.Space == 0)
                    Monitor.Wait(pktTxQ);

                if ((!wakeup) && (pktTxQ.Space > 0))
                {
                    pktTxQ.Enqueue(pkt);
                    result = pkt.Length;
                    pkt    = null;
                    // notify event to protocol
                    Dispatcher.BeginInvoke(ProtocolTask, EventCode.NewPacket);
                }
            }

            return result;
        }

        /// <summary>
        /// Receives a packet
        /// </summary>
        /// <param name="pkt"></param>
        public int RecvPacket(byte[] pkt, int offset, out int count) 
        {
            byte[] tmp;
            int    result;

            count = 0;
            result = RecvPacket(out tmp);

            if (result > 0)
            {
                count = tmp.Length;
                Array.Copy(tmp, 0, pkt, offset, count);
            }

            return result;
        }

        /// <summary>
        /// Receives a packet. This is a blocking API but will be interrupted if the link goes down 
        /// (maximum number of retransmissions excedded). In that case, LinkError event is raised 
        /// in addition to return control to the application.
        /// </summary>
        /// <param name="pkt"></param>
        /// <returns></returns>
        public int RecvPacket(out byte[] pkt)
        {
            int result = 0;
            pkt = null;

            lock (pktRxQ)
            {
                if (pktRxQ.Count == 0)
                    Monitor.Wait(pktRxQ);

                if ((!wakeup) && (pktRxQ.Count > 0))
                {
                    // explanation: it is not clear that count > 0, becuase IsLinkDown could be false
                    // at the time the last consumers adquires the lock
                    pktRxQ.Dequeue(out pkt);
                    Debug.Assert(pkt != null);
                    Debug.Assert(pkt.Length > 0);
                    result = pkt.Length;
                }
            }

            return result;
        }

        /// <summary>
        /// Try again after a link down
        /// </summary>
        public void WakeUpSuspendedThreads()
        {
            // this lock is necesary because could be multiple producers calling TryAgain()
            // concurrently
            lock (evsync)
            {
                if (!wakeup)
                {
                    wakeup = true;

                    // Wake up the aplications (blocked producers and consumers)
                    lock (pktTxQ)
                    {
                        Monitor.PulseAll(pktTxQ);
                    }
                    lock (pktRxQ)
                    {
                        Monitor.PulseAll(pktRxQ);
                    }
                }
            }
        }
        #endregion

        #region Properties
        public int           PacketsToWrite { get { return pktTxQ.Space; } }
        public int           PacketsToRead  { get { return pktRxQ.Count; } }
        public ITrafficStats TxStats        { get { return txstats;      } }
        public ITrafficStats RxStats        { get { return rxstats;      } }
        

        /// <summary>
        /// Input stream
        /// </summary>
        public Stream        In             { get; private set; }

        /// <summary>
        /// Output stream
        /// </summary>
        public Stream        Out            { get; private set; }

        /// <summary>
        /// Maximum Transfer Unit. Maximun amount of data that is possible to send in one packet 
        /// each time
        /// </summary>
        public const int MTU = 10;

        /// <summary>
        /// Number of received packets unfortunately discarded due to overflow in receive buffer. 
        /// To avoid this situation
        /// </summary>
        public long RecvPacketDroped        { get; private set; }
        #endregion

        #region API for derived clases
        protected enum EventCode 
        { 
            NewPacket,
            NewFrame, 
            TxTimeOut
        }

        // API for Protocol: put packet in   pktRxQ queue
        protected void PutPacket(ref byte[] pkt)
        {
            // update traffic stats
            rxstats.NewPacket(pkt.Length * 8);

            lock (pktRxQ)
            {
                if (pktRxQ.Space > 0)
                {
                    pktRxQ.Enqueue(pkt);
                    Monitor.Pulse(pktRxQ);
                }
                else
                {
                    RecvPacketDroped++;
                    //Debug.Assert(false); // commentar eso
                }
            }

            pkt = null;
        }

        protected void PutPacket(/**/byte[] pkt, int offset, int count)
        {
            // update traffic stats
            rxstats.NewPacket(count * 8);
            
            byte[] f = new byte[count];
            Array.Copy(pkt, offset, f, 0, count);

            lock (pktRxQ)
            {
                if (pktRxQ.Space > 0)
                {
                    pktRxQ.Enqueue(f);
                    Monitor.Pulse(pktRxQ);
                }
                else
                {
                    RecvPacketDroped++;
                    //Debug.Assert(false); // commentar eso
                }
            }
        }


        // API for Protocol: get packet from pktTxQ queue
        protected void GetPacket(out byte[] pkt)
        {

            pktTxQ.Dequeue(out pkt);

            lock (pktTxQ)
            {
                Monitor.Pulse(pktTxQ);
            }

            // update traffic stats
            txstats.NewPacket(pkt.Length * 8);
        }

        // API for Protocol: put frame to frmTxQ queue
        protected void PutFrame(ref byte[] frm)
        {   
            // Non-blocking, if there is no space on queue, increment a specific counter 
            // and exit immediately
            lock (frmTxQ)
            {
                if (frmTxQ.Space > 0)
                {
                    frmTxQ.Enqueue(frm);
                    Monitor.Pulse(frmTxQ);
                }
                // else it is lost
            }

            frm = null;
        }


        // API for Protocol: obtains a frame
        protected void GetFrame(out byte[] frm)
        {
            lock (frmRxQ)
            {
                frmRxQ.Dequeue(out frm);
            }
            Debug.Assert(frm != null);
            Debug.Assert(frm.Length > 0);
        }

        // Overriden in a derived class
        protected abstract void ProtocolTask(object reason);
        #endregion

        #region COBS Encoder/Decoder Threads
        // COBS (Consistent Overhead Byte-Stuffing) Encoder Thread
        void FrameEncoder()
        {
            byte[] f = null;
            byte   b;
            
            while (true)
            {
                lock (frmTxQ)
                {
                    if (frmTxQ.Count == 0)
                        Monitor.Wait(frmTxQ);

                    frmTxQ.Dequeue(out f);
                    Monitor.Pulse(frmTxQ);
                }

                Debug.Assert(f != null);
                Debug.Assert(f.Length > 0);//???????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????

                codec.PutFrame(f);
                while (codec.EncoderBytes > 0)
                {
                    codec.GetByte(out b);
                    Out.WriteByte(/**/b);
                }
            }
        }

        // COBS (Consistent Overhead Byte-Stuffing) Decoder Thead
        void FrameDecoder()
        {
            byte[] f;
            byte[] b = new byte[1];

            while (true)
            {
                In.Read(b, 0, 1);
                codec.PutByte(b[0]);

                if (codec.FrameAvailable)
                {
                    // create a buffer
                    
                    codec.GetFrame(out f);

                    Debug.Assert(f != null);
                    Debug.Assert(f.Length > 0);
                        
                    lock (frmRxQ)
                    {
                        if (frmRxQ.Space > 0)
                        {
                            frmRxQ.Enqueue(/**/f);
                            Dispatcher.BeginInvoke(ProtocolTask, EventCode.NewFrame);
                        }
                        // else it is lost
                    }
                }
            }
        }
        #endregion
    }
}
